package com.tiger.job;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.tiger.entity.trade.CoinThumb;
import com.tiger.entity.trade.ExchangeTrade;
import com.tiger.entity.trade.TradePlate;
import com.tiger.entity.trade.TradePlateItem;
import com.tiger.entity.util.MarketConstants;
import com.tiger.entity.util.MarketEnums;
import com.tiger.entity.vo.ExchangeBuy1Sell1Vo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;

@Slf4j
@Component
/**
 * 交易推送
 */
public class ExchangePushJob {
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    private Map<String, List<ExchangeTrade>> tradesQueue;
    private Map<String, List<TradePlate>> plateQueue ;
    private Map<String, List<CoinThumb>> thumbQueue;

    private Map<String, TradePlate> plateLastBuy; // 最后一次推送的盘口，仅仅是为了虚拟推送二设立的
    private Map<String, TradePlate> plateLastSell; // 最后一次推送的盘口，仅仅是为了虚拟推送二设立的

    private Map<String, BigDecimal> plateBuy1; // 买1
    private Map<String, BigDecimal> plateSell1; // 卖1


    public ExchangePushJob() {
        this.tradesQueue = new HashMap<>();
        this. plateQueue = new HashMap<>();
        this.thumbQueue = new HashMap<>();

        this.plateLastBuy = new HashMap<>();
        this.plateLastSell = new HashMap<>();

        this.plateBuy1 = new HashMap<>();
        this.plateSell1 = new HashMap<>();

    }

    public void addTrades(String symbol, List<ExchangeTrade> trades) {
        List<ExchangeTrade> list = tradesQueue.get(symbol);
        if (list == null) {
            list = new ArrayList<>();
            tradesQueue.put(symbol, list);
        }
        synchronized (list) {
            list.addAll(trades);
        }
    }

    public void addPlates(String symbol, TradePlate plate) {
        List<TradePlate> list = this.plateQueue.get(symbol);
        if (list == null) {
            list = new ArrayList<>();
            this.plateQueue.put(symbol, list);
        }
        synchronized (list) {
            list.add(plate);
        }
        if (plate.getDirection() == MarketEnums.OrderDirection.Buy) {
            BigDecimal highestPrice = plate.getHighestPrice();
            this.plateBuy1.put(symbol,highestPrice);
            // 更新最新盘口
            synchronized (plateLastBuy) {
                plateLastBuy.put(symbol, plate);
            }

        }
        if (plate.getDirection() == MarketEnums.OrderDirection.Sell) {
            BigDecimal lowestPrice = plate.getLowestPrice();

            this.plateSell1.put(symbol,lowestPrice);
            // 更新最新盘口
            synchronized (plateLastSell) {
                plateLastSell.put(symbol, plate);
            }
        }

    }

    public void addThumb(String symbol, CoinThumb thumb) {
        List<CoinThumb> list = thumbQueue.get(symbol);
        if (list == null) {
            list = new ArrayList<>();
            thumbQueue.put(symbol, list);
        }
        synchronized (list) {
            list.add(thumb);
        }
    }

    /**
     * 推送成交
     */
    @Scheduled(fixedRate = 300)
    public void pushTrade() {
        for (Map.Entry<String, List<ExchangeTrade>> entry : tradesQueue.entrySet()) {
            String symbol = entry.getKey();
            List<ExchangeTrade> trades = entry.getValue();
            if (trades.size() > 0) {
                synchronized (trades) {
                    messagingTemplate.convertAndSend("/topic/market/trade/" + symbol, trades);
                    trades.clear();
                }
            }
        }
    }

    /**
     * 盘口 深度推送
     */
    @Scheduled(fixedDelay = 500)
    public void pushPlate() {
        for (Map.Entry<String, List<TradePlate>> entry : plateQueue.entrySet()) {
            String symbol = entry.getKey();
            List<TradePlate> plates = entry.getValue();
            if (plates.size() > 0) {
                boolean hasPushAskPlate = false;
                boolean hasPushBidPlate = false;
                synchronized (plates) {
                    for (TradePlate plate : plates) {
                        if (plate.getDirection() == MarketEnums.OrderDirection.Buy && !hasPushBidPlate) {
                            hasPushBidPlate = true;

                        } else if (plate.getDirection() == MarketEnums.OrderDirection.Sell && !hasPushAskPlate) {
                            hasPushAskPlate = true;
                        } else {
                            continue;
                        }
                        //websocket推送盘口信息

                        messagingTemplate.convertAndSend("/topic/market/trade-plate/" + symbol, plate.toJSON(24));
                        //websocket推送深度信息
                        messagingTemplate.convertAndSend("/topic/market/trade-depth/" + symbol, plate.toJSON(50));
                        //netty推送
                        // nettyHandler.handlePlate(symbol, plate);
                    }

                    plates.clear();
                }
            } else {
                // 不管盘口有没有变化，都推送一下数据，显得盘口交易很活跃的样子(这里获取到的盘口有可能是买盘，也可能是卖盘)
                TradePlate plateBuy = plateLastBuy.get(symbol);
                TradePlate plateSell = plateLastSell.get(symbol);
                if (plateBuy != null) {
                    //websocket推送盘口信息
                    messagingTemplate.convertAndSend("/topic/market/trade-plate/" + symbol, plateBuy.toJSON(24));
                    //websocket推送深度信息
                    messagingTemplate.convertAndSend("/topic/market/trade-depth/" + symbol, plateBuy.toJSON(50));
                    //netty推送
                    // nettyHandler.handlePlate(symbol, plateBuy);
                }
                if (plateSell != null) {
                    //websocket推送盘口信息
                    messagingTemplate.convertAndSend("/topic/market/trade-plate/" + symbol, plateSell.toJSON(24));
                    //websocket推送深度信息
                    messagingTemplate.convertAndSend("/topic/market/trade-depth/" + symbol, plateSell.toJSON(50));
                    //netty推送
                    // nettyHandler.handlePlate(symbol, plateSell);
                }
            }
        }
    }

    /**
     * 推送缩略信息
     */
    @Scheduled(fixedRate = 300)
    public void pushThumb() {
        Iterator<Map.Entry<String, List<CoinThumb>>> entryIterator = thumbQueue.entrySet().iterator();
        //log.info("thumbQueue::::"+ JSON.toJSONString(thumbQueue));
        while (entryIterator.hasNext()) {
            Map.Entry<String, List<CoinThumb>> entry = entryIterator.next();
            String symbol = entry.getKey();
            List<CoinThumb> thumbs = entry.getValue();
            if (thumbs.size() > 0) {
                synchronized (thumbs) {
                    messagingTemplate.convertAndSend("/topic/market/thumb", thumbs.get(thumbs.size() - 1));
                    thumbs.clear();
                }
                return;
            }
        }

    }
    @Scheduled(fixedRate = 1000*15)
    public  void pushBuy1Sell1(){
        // 后期可优化成买1卖1没更新就不推送
        Iterator<Map.Entry<String, BigDecimal>> iterator = plateBuy1.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, BigDecimal> entry = iterator.next();
            String symbol = entry.getKey();
            BigDecimal buy = entry.getValue();
            BigDecimal sell = plateSell1.get(symbol);
            ExchangeBuy1Sell1Vo build = ExchangeBuy1Sell1Vo.builder().buy1(buy).sell1(sell).symbol(symbol).build();
            rabbitTemplate.convertAndSend(MarketConstants.EXCHANGE_BUY1_SELL1_PLATE, JSONUtil.toJsonStr(build));
        }
    }
}
